package com.atguigu.datastream.day07;

import com.atguigu.datastream.bean.Event;
import com.atguigu.datastream.test.day03.Flink_05_Source_useDefault;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * ClassName: Flink02_SplitStreamByFilter
 * Package: com.atguigu.day07
 * Description:
 *            将一条流根据不同的用户拆分成多条流
 * @Author ChenJun
 * @Create 2023/4/14 16:33
 * @Version 1.0
 */
public class Flink02_SplitStreamByFilter {
    public static void main(String[] args) throws Exception {

        //1. 创建流的运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        //2. 获取数据
        DataStreamSource<Event> streamSource = env.addSource(new Flink_05_Source_useDefault.ClickSource());

        //3. 将一条流根据不同的用户拆分成多条流
        SingleOutputStreamOperator<Event> songsongDStream = streamSource.filter(new FilterFunction<Event>() {
            @Override
            public boolean filter(Event value) throws Exception {
                return "Songsong".equals(value.user);
            }
        });

        SingleOutputStreamOperator<Event> bingbingDStream = streamSource.filter(new FilterFunction<Event>() {
            @Override
            public boolean filter(Event value) throws Exception {
                return "Bingbing".equals(value.user);
            }
        });

        songsongDStream.print("songsong:");
        bingbingDStream.print("bingbing:");

        env.execute();
    }
}
